package com.jusda.message;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.jusda.message.entity.UserScore;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;

/**
 * @author:y022 23/04/2022 12:38
 * @description:
 */
@Configuration
@Slf4j
@AllArgsConstructor
public class Consumer {
    private final ObjectMapper objectMapper;

    @Bean
    public DefaultMQPushConsumer pushConsumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("rocketmq-service");
        try {
            consumer.setNamesrvAddr("121.5.246.4:9876");
            consumer.subscribe("rocketmq-service", "a || b");
            consumer.setConsumeMessageBatchMaxSize(5);
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @SneakyThrows
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                    context.setAutoCommit(true);
                    for (MessageExt msg : msgs) {
                        log.info("当前消费线程:{},当前分区Id:{},消息体:{}", Thread.currentThread().getName(), msg.getQueueId(), objectMapper.readValue(msg.getBody(), UserScore.class));
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });
            consumer.start();
        } catch (Exception e) {
            log.info("消息消费失败!");
        }
        return consumer;
    }
}
